package com.hugh.commom.selector;

import com.hugh.commom.netty.client.NettyClientOpen;
import com.hugh.commom.zk.client.entity.ZkNodeEntity;
import com.hugh.commom.zk.client.ZkRpcHandle;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: Hugh
 * @Date: 2020/6/22
 */
@Component(value = "channelSelect")
public class ChannelSelect {

    @Autowired
    private ZkRpcHandle zkUtil;

    private Map<String, List<ZkNodeEntity>> ipPortMap = new ConcurrentHashMap<>();

    private Map<String, List<ChannelFutureEntity>> channelMap = new ConcurrentHashMap<>();

    /**
     *  key:srvice
     *  val:ip,ip
     * @param ipPortMap
     * @throws Exception
     */
    public void initChannelSelect(Map<String, List<ZkNodeEntity>> ipPortMap) throws Exception {
        if(Objects.isNull(ipPortMap) || ipPortMap.size() <= 0){
            return;
        }
        this.ipPortMap = ipPortMap;
        channelMap = new HashMap<>();
        for(Map.Entry<String, List<ZkNodeEntity>> entry : ipPortMap.entrySet()){

            List<ChannelFutureEntity> channelFutures = new ArrayList<>(entry.getValue().size());
            for(ZkNodeEntity znode : entry.getValue()){
                Bootstrap bootstrap = new NettyClientOpen().doOpen();
                channelFutures.add(
                       ChannelFutureEntity.builder()
                                .ip(znode.getIp())
                                .port(znode.getPort())
                                .bootstrap(bootstrap)
                                .service(entry.getKey())
                                .rt(znode.getRt())
                                .build());
            }
            channelMap.put(entry.getKey(), channelFutures);
        }
    }

    /**
     * 更新
     * @param serviceName 节点名字
     * @param zkNodeList zkNodeList
     * @throws Exception
     */
    public void updateChannelSelect(String serviceName, List<ZkNodeEntity> zkNodeList) throws Exception {
        if(ipPortMap.containsKey(serviceName)){
                //移除老的
            ipPortMap.remove(serviceName);
        }
        if(!CollectionUtils.isEmpty(zkNodeList)) {
            //构建新的
            buildNewChannel(serviceName, zkNodeList);
            this.ipPortMap.put(serviceName, zkNodeList);
        }
    }

    private void buildNewChannel(String key, List<ZkNodeEntity> ipPorts) throws Exception {
        if(ipPorts.isEmpty()){
            return;
        }
        List<ChannelFutureEntity> channelFutures = channelMap.get(key);
        if(CollectionUtils.isEmpty(channelFutures)){
            channelFutures = new ArrayList<>();
        }
        for(ZkNodeEntity znode : ipPorts){
            Bootstrap bootstrap = new NettyClientOpen().doOpen();
            channelFutures.add(
                    ChannelFutureEntity.builder()
                            .ip(znode.getIp())
                            .port(znode.getPort())
                            .bootstrap(bootstrap)
                            .service(key)
                            .rt(znode.getRt())
                            .build());
        }
        channelMap.put(key, channelFutures);
    }

    public ChannelFutureEntity select(String service) throws Exception {
        if(!channelMap.containsKey(service)){
            throw new Exception("select no provider");
        }
        List<ChannelFutureEntity> channelFutureEntities = channelMap.get(service);
        Optional<ChannelFutureEntity> first = channelFutureEntities.stream().sorted(Comparator.comparing(ChannelFutureEntity::getRt)).findFirst();
        //实际建立连接
        if(first.isEmpty()){
            throw new Exception("select no provider");
        }
        ChannelFutureEntity channelFutureEntity = first.get();
        Bootstrap bootstrap = channelFutureEntity.getBootstrap();
        ChannelFuture sync = bootstrap.connect(channelFutureEntity.getIp(), channelFutureEntity.getPort()).sync();
        channelFutureEntity.setChannelFuture(sync);
        return channelFutureEntity;
    }
}
